package com.flinksql.test;

import com.flinksql.bean.WaterSensor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author: Lin
 * @create: 2021-06-16 10:21
 * @description: FlinkSQL，从kafka读取数据，并写入到kafka
 **/
public class FlinkSQL_Test3 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        //1.创建表的执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.注册sourcetable
        tableEnv.executeSql("CREATE TABLE source_sensor (" +
                "  `id` STRING," +
                "  `ts` BIGINT," +
                "  `vc` INT" +
                ") WITH (" +
                "  'connector' = 'kafka'," +
                "  'topic' = 'topic_source_sensor'," +
                "  'properties.bootstrap.servers' = 'hadoop102:9092'," +
                "  'properties.group.id' = 'testGroup'," +
                "  'scan.startup.mode' = 'latest-offset'," +
                "  'format' = 'csv'" +
                ")");

        tableEnv.executeSql("CREATE TABLE sink_sensor (" +
                "  `id` STRING," +
                "  `ts` BIGINT," +
                "  `vc` INT" +
                ") WITH (" +
                "  'connector' = 'kafka'," +
                "  'topic' = 'topic_sink_sensor'," +
                "  'properties.bootstrap.servers' = 'hadoop102:9092'," +
                "  'format' = 'json'" +
                ")");

        //3.使用sql语句从source查询后写入到sinktable
        tableEnv.executeSql("insert into sink_sensor select * from source_sensor");

    }
}
